import com.sun.org.slf4j.internal.Logger;
import com.sun.org.slf4j.internal.LoggerFactory;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.etcd.jetcd.kv.TxnResponse;
import io.etcd.jetcd.op.Cmp;
import io.etcd.jetcd.op.CmpTarget;
import io.etcd.jetcd.op.Op;
import io.etcd.jetcd.options.PutOption;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.ApplicationProtocolNames;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.File;
import java.nio.charset.Charset;
import java.util.List;

public class ETCDUtilL {
    public static Logger log = LoggerFactory.getLogger(ETCDUtilL.class);
    //etcl客户端链接
    private static Client client = null;
    private static final String endpoints = "https://192.168.1.152:2379,https://192.168.1.153:2379,https://192.168.1.155:2379";
    public static synchronized Client getEtclClient() throws Exception {
        File CA = new File("C:\\JAVAProject\\SSLTest\\src\\main\\resources\\ca-2.crt");

        if(null == client){

            ApplicationProtocolConfig apn = new ApplicationProtocolConfig(ApplicationProtocolConfig.Protocol.ALPN,
                    ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
                    ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
                    ApplicationProtocolNames.HTTP_2,
                    ApplicationProtocolNames.HTTP_1_1);

//            SslContext sslContext = SslContextBuilder.forClient().applicationProtocolConfig(apn).trustManager(InsecureTrustManagerFactory.INSTANCE).build();
            SslContext sslContext = SslContextBuilder.forClient().applicationProtocolConfig(apn).trustManager(CA).build();
            client = Client.builder().sslContext(sslContext).endpoints(endpoints.split(",")).build();

        }
        return client;
    }

    /**
     * 根据指定的配置名称获取对应的value
     * @param key 配置项
     * @return
     * @throws Exception
     */
    public static String getEtcdValueByKey(String key) throws Exception {
        List<KeyValue> kvs = ETCDUtilL.getEtclClient().getKVClient().get(ByteSequence.from(key, Charset.defaultCharset())).get().getKvs();
        if(kvs.size()>0){
            ByteSequence byteSequence = kvs.get(0).getValue();
            String value = byteSequence.toString(Charset.defaultCharset());
            return value;
        }
        else {
            return null;
        }
    }

    /**
     * 新增或者修改指定的配置
     * @param key
     * @param value
     * @return
     */
    public static void putEtcdValueByKey(String key,String value) throws Exception{
        ETCDUtilL.getEtclClient().getKVClient().put(ByteSequence.from(key,Charset.defaultCharset()),ByteSequence.from(value,Charset.defaultCharset()));
    }

    /**
     * 删除指定的配置
     * @param key
     * @return
     */
    public static void deleteEtcdValueByKey(String key) throws Exception {
        ETCDUtilL.getEtclClient().getKVClient().delete(ByteSequence.from(key,Charset.defaultCharset()));
    }

    /**
     * etcd的监听，监听指定的key，当key 发生变化后，监听自动感知到变化。
     * @param key 指定需要监听的key
     */
    public static void initListen(String key, SourceFunction.SourceContext<String> ctx) {
        try {
            //加载配置
            getConfig(ETCDUtilL.getEtclClient().getKVClient().get(ByteSequence.from(key,Charset.defaultCharset())).get().getKvs());
            new Thread(() -> {
                Watch.Watcher watcher = null;
                try {
                    MyConsumer consumer = new MyConsumer();
                    MyRunable myRunable = new MyRunable();
                        watcher = ETCDUtilL.getEtclClient().getWatchClient().watch(ByteSequence.from(key, Charset.defaultCharset()), Watch.listener(consumer,myRunable));
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }).start();
        } catch (Exception e) {
            System.out.println("etcd listen start cause Exception:{"+e+"}");
        }

    }
    private static String getConfig(List<KeyValue> kvs){
        if(kvs.size()>0){
            String config = kvs.get(0).getKey().toString(Charset.defaultCharset());
            String value = kvs.get(0).getValue().toString(Charset.defaultCharset());
            System.out.println("etcd 's config 's config key is :{"+config+"},value is:{"+value+"}");
            return value;
        }
        else {
            return null;
        }
    }

    /**
     * 将字符串转为客户端所需的ByteSequence实例
     *
     * @param val
     * @return
     */
    private static ByteSequence bytesOf(String val) {
        return ByteSequence.from(val, Charset.defaultCharset());
    }

    public static boolean transaction(String conditionKey, String conditionValue, String updateValue) throws Exception {
        // 将三个String型的入参全部转成ByteSequence类型
        ByteSequence conKey = bytesOf(conditionKey);
        ByteSequence conValue = bytesOf(conditionValue);
        ByteSequence bsUpdateValue = bytesOf(updateValue);

        // 是否相等的比较
        Cmp cmp = new Cmp(conKey, Cmp.Op.EQUAL, CmpTarget.value(conValue));

        // 执行事务
        TxnResponse txnResponse = getEtclClient()
                .getKVClient()
                .txn()
                .If(cmp)
                .Then(Op.put(conKey, bsUpdateValue, PutOption.DEFAULT))
                .commit()
                .get();
        System.out.println(txnResponse.getPutResponses());
        // 如果操作成功，isSucceeded方法会返回true，并且getPutResponse也有内容
        return txnResponse.isSucceeded() && !txnResponse.getPutResponses().isEmpty();
    }

}
